package ar.edu.itba.pod.legajo47573;

import java.rmi.AlreadyBoundException;
import java.rmi.NotBoundException;
import java.rmi.RemoteException;
import java.rmi.registry.LocateRegistry;
import java.rmi.registry.Registry;
import java.rmi.server.UnicastRemoteObject;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;

import ar.edu.itba.balance.api.AgentsBalancer;
import ar.edu.itba.balance.api.AgentsTransfer;
import ar.edu.itba.balance.api.NodeAgent;
import ar.edu.itba.balance.api.NotCoordinatorException;
import ar.edu.itba.node.Node;
import ar.edu.itba.node.NodeInformation;

@SuppressWarnings("serial")
public class AgentsBalancerImpl extends UnicastRemoteObject implements AgentsBalancer{ 

	private MyNode currentNode;
	private ConcurrentMap<NodeInformation, Long> updates = new ConcurrentHashMap<NodeInformation, Long>();
	private ConcurrentMap<NodeInformation, Long> coordinators = new ConcurrentHashMap<NodeInformation, Long>();
	private Thread timeout;
	/*Cuando está en una elección que no se balancee*/
	private CountDownLatch latch = null;
	private CountDownLatch balancerLatch = null;
	//private AtomicBoolean balancing = new AtomicBoolean(false);
	private AtomicBoolean shutdown = new AtomicBoolean(false);

	/*Empiezan todos siendo posibles candidatos a coordinador*/
	private AtomicBoolean coordinator = new AtomicBoolean(true);
	private AtomicBoolean enteredShutdown = new AtomicBoolean(false);
	private ConcurrentMap<NodeInformation, Integer> agentsNumber = new ConcurrentHashMap<NodeInformation, Integer>();
	private ConcurrentMap<NodeInformation, Integer> agentsToMove = new ConcurrentHashMap<NodeInformation, Integer>();
	private List<NodeAgent> agentsToRedistribute = new ArrayList<NodeAgent>();

	public AgentsBalancerImpl(MyNode currentNode) throws RemoteException {
		super();
		this.currentNode = currentNode;
		try {
			currentNode.getRegistry().bind(Node.AGENTS_BALANCER, this);
		} catch (AlreadyBoundException e) {
			e.printStackTrace();
		}
	}

	public void bullyElection(NodeInformation node, long timestamp){
		System.out.println("Bully Election");
		if(updates.get(node) == null || updates.get(node) < timestamp){
			coordinator.set(true);
			latch = new CountDownLatch(1);
			updates.put(node, timestamp);
			if(currentNode.getInfo().id().compareTo(node.id()) < 0 ){
				coordinator.set(false);
				if(coordinator.get() != false)
					election(node, timestamp);
			}else if (currentNode.getInfo().id().compareTo(node.id()) > 0 ){
				AgentsBalancer balancerStub = (AgentsBalancer)Registries.getRemoteObject(currentNode, node, Node.AGENTS_BALANCER);
				try {
					balancerStub.bullyOk(node);
				} catch (RemoteException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
				election(currentNode.getInfo(), System.currentTimeMillis());
				waiting();
			}
			else{
				election(currentNode.getInfo(), timestamp);
				waiting();
			}
		}
	}

	private void election(final NodeInformation node, final long newTimestamp){
		Thread election = new Thread(){
			public void run(){
				try {
					for(NodeInformation it : currentNode.getCluster().connectedNodes()){
						if(!it.equals(currentNode.getInfo()) && !it.equals(node)){
							AgentsBalancer stub = (AgentsBalancer)Registries.getRemoteObject(currentNode, it, Node.AGENTS_BALANCER);
							stub.bullyElection(currentNode.getInfo(), newTimestamp);
						}
					}
				} catch (RemoteException e) {
					e.printStackTrace();
				}
			}
		};
		election.start();
	}

	private void waiting(){
		if(timeout != null && timeout.isAlive()){
			timeout.interrupt();
		}
		Thread timeout = new Thread(){

			public void run() {
				try {
					Thread.sleep(2000);
					if(coordinator.get() == true){
						bullyCoordinator(currentNode.getInfo(), System.currentTimeMillis());
					}
				} catch (InterruptedException e) {
					e.printStackTrace();
				} catch (RemoteException e) {
					e.printStackTrace();
				}
			};
		};
		timeout.start();
	}

	@Override
	public void bullyOk(NodeInformation node) throws RemoteException {
		this.coordinator.set(false);
	}

	@Override
	public void bullyCoordinator(NodeInformation node, long timestamp) throws RemoteException {
		System.out.println("ENTRO AL COORDINATOR");
		if(coordinators.get(node) == null || coordinators.get(node) < timestamp){
		//	balancing.set(true);
			balancerLatch = new CountDownLatch(1);
			coordinators.put(node, timestamp);
			System.out.println("BULLY COORDINATOR " + node);

			/*Seteo en mi nodo quién es el coordinador y después hago el broadcast*/
			currentNode.setCoordinator(node);

			/*Hago el broadcast*/
			Set<NodeInformation> nodes = new HashSet<NodeInformation>(currentNode.getCluster().connectedNodes());

			for(NodeInformation it : nodes){
				AgentsBalancer balancerStub;
				try {
					Registry balancerRegistry = LocateRegistry.getRegistry(it.host(), it.port());
					balancerStub = (AgentsBalancer)balancerRegistry.lookup(Node.AGENTS_BALANCER);
					balancerStub.bullyCoordinator(node, timestamp);
				} catch (NotBoundException e) {
					e.printStackTrace();
				}
			}


			/*Si soy el nuevo coordinador cargo el mapa de nodos y número de agentes*/
			if(currentNode.getInfo().equals(node)){
				Set<NodeInformation> nodos = currentNode.getCluster().connectedNodes();
				nodos.add(currentNode.getInfo());
				for(NodeInformation it : nodos){
					AgentsTransfer stub = (AgentsTransfer)Registries.getRemoteObject(currentNode, it, Node.AGENTS_TRANSFER);
					System.out.println(it);
					System.out.println(stub.getNumberOfAgents());
					agentsNumber.put(it, stub.getNumberOfAgents());
				}
				latch.countDown();

				//	if(enteredShutdown.get() == true)
				balancer();
				//	enteredShutdown.set(false);
			}
			balancerLatch.countDown();
		//	balancing.set(false);
		}
	}

	public void balancer() throws RemoteException{
		
		Set<NodeInformation> nodesaux = currentNode.getCluster().connectedNodes();
		for(NodeInformation it : nodesaux){
			if(!agentsNumber.containsKey(it))
				agentsNumber.put(it, 0);
		}
		System.out.println("BALANCER");
		int minaux = Integer.MAX_VALUE;
		int maxaux = -1;
		Iterator<Entry<NodeInformation, Integer>> itaux = agentsNumber.entrySet().iterator();
		while(itaux.hasNext()){
			Entry<NodeInformation, Integer> entry = (Entry<NodeInformation, Integer>) itaux.next();
			int aux = entry.getValue();
			if(aux < minaux){
				minaux = aux;
			}
			if(aux > maxaux){
				maxaux = aux;
			}
		}

		agentsToMove.clear();

		System.out.println(maxaux);
		System.out.println(minaux);

		int min = Integer.MAX_VALUE;
		int max = -1;
		NodeInformation entryMin = null;
		NodeInformation entryMax = null;

		if(maxaux - minaux > 1){
			if(latch != null){
				try {
					latch.await();
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
			System.out.println("ACA NO ENTRO");
			while(max - min <= 1){
				Iterator<Entry<NodeInformation, Integer>> it3 = agentsNumber.entrySet().iterator();
				while(it3.hasNext()){
					Entry<NodeInformation, Integer> entry = (Entry<NodeInformation, Integer>) it3.next();
					int aux = entry.getValue();
					if(aux < min){
						min = aux;
						entryMin = entry.getKey();
					}
					if(aux > max){
						max = aux;
						entryMax = entry.getKey();
					}
				}
				int average = (min + max)/2;
				int others = (min + max) % 2;
				Integer prevValueMin = agentsToMove.get(entryMin);
				Integer prevValueMax = agentsToMove.get(entryMax);
				if(prevValueMin == null)
					prevValueMin = 0;
				if(prevValueMax == null)
					prevValueMax = 0;
				agentsToMove.put(entryMin, prevValueMin + (average - agentsNumber.get(entryMin)));
				agentsToMove.put(entryMax, prevValueMax + ((average + others) - agentsNumber.get(entryMax)));
				agentsNumber.put(entryMin, average);
				agentsNumber.put(entryMax, average + others);
			}

			Iterator<Entry<NodeInformation, Integer>> move = agentsToMove.entrySet().iterator();

			while(move.hasNext()){
				Entry<NodeInformation, Integer> entry = (Entry<NodeInformation, Integer>) move.next();
				System.out.println("valores a mover " + entry.getValue());
				/*tomo agentes*/
				if(entry.getValue() < 0){
					System.out.println("HAY MENORES A CERO");
					AgentsTransfer stub = (AgentsTransfer)Registries.getRemoteObject(currentNode, entry.getKey(), Node.AGENTS_TRANSFER);
					try {
						agentsToRedistribute.addAll((stub.stopAndGet(Math.abs(entry.getValue()))));
					} catch (RemoteException e) {
						e.printStackTrace();
					}
				}
			}

			int limInf=0;
			int limSup=0;

			Iterator<Entry<NodeInformation, Integer>> move2 = agentsToMove.entrySet().iterator();

			while(move2.hasNext()){
				Entry<NodeInformation, Integer> entry = (Entry<NodeInformation, Integer>) move2.next();
				/*tomo agentes*/
				System.out.println(entry.getValue());
				if(entry.getValue() > 0){
					System.out.println("HAY MAYORES A CERO");
					limSup = entry.getValue();
					AgentsTransfer stub = (AgentsTransfer)Registries.getRemoteObject(currentNode, entry.getKey(), Node.AGENTS_TRANSFER);
					try {
						System.out.println(agentsToRedistribute.subList(limInf, limSup));
						stub.runAgentsOnNode(new ArrayList<NodeAgent>(agentsToRedistribute.subList(limInf, limSup)));
						limInf = limSup;
					} catch (RemoteException e) {
						e.printStackTrace();
					}
				}
			}

			Set<NodeInformation> nodes = new HashSet<NodeInformation>(currentNode.getCluster().connectedNodes());
			nodes.add(currentNode.getInfo());
			System.out.println("AGENTES CORRIENDO " + currentNode.getSimulation().agentsRunning());

			for(NodeInformation it1 : nodes){
				System.out.println("Soy el nodo en balancer" + it1 + " estoy ejecutando " + 
						agentsNumber.get(it1) + " agentes");
			}
		}
		System.out.println("SALGO DEL BALANCER");

	}


	@Override
	public void shutdown(List<NodeAgent> agents) throws RemoteException, NotCoordinatorException {
		if(!currentNode.getCoordinator().equals(currentNode.getInfo())){
			throw new NotCoordinatorException(currentNode.getCoordinator());
		}

		enteredShutdown.set(true);
		synchronized(AgentsBalancer.class){
		NodeInformation node = agents.get(0).node();

		agentsNumber.remove(node);

		for(NodeAgent it : agents){
			shutdown.set(true);
			addAgentToCluster(it);
		}
		}
	}

	@Override
	public void addAgentToCluster(NodeAgent agent) throws RemoteException, NotCoordinatorException {

		if(balancerLatch != null){
			try {
				balancerLatch.await();
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
		
		if(!currentNode.getCoordinator().equals(currentNode.getInfo())){
			throw new NotCoordinatorException(currentNode.getCoordinator());
		}

		synchronized(AgentsBalancer.class){
			Set<NodeInformation> nodes = new HashSet<NodeInformation>(currentNode.getCluster().connectedNodes());
			nodes.add(currentNode.getInfo());
			/*Borro al nodo de la lista porque quiere decir que viene de un shutdown*/
			if(shutdown.get() == true){
				System.out.println("Es SHUTDOWN");
				nodes.remove(agent.node());
			}
			System.out.println("Adentro de addAgentToCluster");
			for(NodeInformation it : nodes){
				if(!agentsNumber.containsKey(it)){
					agentsNumber.put(it, 0);
				}
			}
			int min = Integer.MAX_VALUE;
			NodeInformation minNode = null;

			/*Busco el nodo con menos agentes*/
			for(NodeInformation it : nodes){
				if(agentsNumber.get(it) < min){
					minNode = it;
					min = agentsNumber.get(it);
				}
			}

			/*TODO DEBUG*/
			agentsNumber.put(minNode, min+1);
			for(NodeInformation it : nodes){
				System.out.println("Soy el nodo " + it + " estoy ejecutando " + 
						agentsNumber.get(it) + " agentes");
			}
			
			AgentsTransfer stub = (AgentsTransfer)Registries.getRemoteObject(currentNode, minNode, Node.AGENTS_TRANSFER);
			List<NodeAgent> parameter = new ArrayList<NodeAgent>();
			parameter.add(agent);
			stub.runAgentsOnNode(parameter);

			shutdown.set(false);
		}
	}

}
